// 2015年版权所有etcd作者
// 
// 根据Apache许可证2.0版（以下简称“许可证”）获得许可；
// 除非遵守许可证，否则不得使用此文件。
// 您可以通过
// 
// www.apache.org/licenses/LICENSE-2.0 
// 
// 除非适用法律要求或书面同意，否则根据许可证分发的软件
// 按“原样”分发，
// 无任何明示或暗示的保证或条件。
// 请参阅许可证，了解管理许可的特定语言和
// 许可证下的限制。

package raft

import (
	"context"
	"errors"

	pb "go.etcd.io/etcd/raft/v3/raftpb"
)

type SnapshotStatus int

const (
	SnapshotFinish  SnapshotStatus = 1
	SnapshotFailure SnapshotStatus = 2
)

var (
	emptyState = pb.HardState{}

	// ErrStopped由已停止的节点上的方法返回。
	ErrStopped = errors.New("raft: stopped")
)

// SoftState提供对日志记录和调试有用的状态。
// 状态是易变的，不需要持久化到WAL。
type SoftState struct {
	Lead      uint64 // 必须使用原子操作才能访问；保持64位对齐。
	RaftState StateType
}

func (a *SoftState) equal(b *SoftState) bool {
	return a.Lead == b.Lead && a.RaftState == b.RaftState
}

// 内嵌了SoftState和HardState，在SofteState中封装了当前集群的Leader节点ID(Lead宇段〉及当前节点的角色（RaftState），
// 在HardState中封装了当前节点的任期号（Term字段〉、当前节点在该任期投票结果（Vote字段）及当前节点的raftLog的己提交位置
type Ready struct {
	// SofteState中封装了当前集群的Leader节点ID(Lead宇段〉及当前节点的角色（RaftState）
	*SoftState
	// HardState中封装了当前节点的任期号（Term字段〉、当前节点在该任期投票结果（Vote字段）及当前节点的raftLog的己提交位置
	pb.HardState
	// 记录了当前节点中等待处理的只读请求。
	ReadStates []ReadState
	// Entry记录是从unstable中读取出来的，上层模块会将其保存到Storage中
	Entries []pb.Entry
	// 待持久化的快照数据，ra企pb.Snapshot中封装了快照数据及相关元数据
	Snapshot pb.Snapshot
	// 己提交、待应用的Entry记录，这些En町记录之前己经保存到了Storage中
	CommittedEntries []pb.Entry
	// 保存了当前节点中等待发送到集群其他节点的Message消息
	Messages []pb.Message
	MustSync bool
}

func isHardStateEqual(a, b pb.HardState) bool {
	return a.Term == b.Term && a.Vote == b.Vote && a.Commit == b.Commit
}

// IsEmptyHardState如果给定的HardState为空，则返回true。
func IsEmptyHardState(st pb.HardState) bool {
	return isHardStateEqual(st, emptyState)
}

// 如果给定快照为空，IsEmptySnap将返回true。
func IsEmptySnap(sp pb.Snapshot) bool {
	return sp.Metadata.Index == 0
}

// 通过newReady（）方法得到Ready实例之后，会通过containsUpdates（）方法检测Ready实例中各个宇段是否为空，
// 从而决定此次创建的Ready实例是否有必要交给上层模块处理进行处理
func (rd Ready) containsUpdates() bool {
	return rd.SoftState != nil || !IsEmptyHardState(rd.HardState) ||	// 检测raft状态是否发生变化
		!IsEmptySnap(rd.Snapshot) || 	// 检测是否有新的快照数据
		len(rd.Entries) > 0 ||	// 检测是否有待持久化的Entry记录
		len(rd.CommittedEntries) > 0 ||	// 检测是否有待应用的Entry记录
		len(rd.Messages) > 0 ||	// 检测是否有待发送的消息
		len(rd.ReadStates) != 0	// 检测是否有处理的只读请求
}

// appliedCursor从就绪中提取客户端已应用的最高索引
// （通过Advance确认就绪后）。如果Ready中没有包含任何信息
// 则返回零。
func (rd Ready) appliedCursor() uint64 {
	if n := len(rd.CommittedEntries); n > 0 {
		return rd.CommittedEntries[n-1].Index
	}
	if index := rd.Snapshot.Metadata.Index; index > 0 {
		return index
	}
	return 0
}

// 在etcd-raft模块中，结构体Node表示集群中的一个节点，
// 它是在结构体raft之上的一层封装，对外提供相对简单的API接口
type Node interface {
	// 是用来推进逻辑时钟的指针，从而推进选举计时器和心跳计时器
	Tick()
	//当选举计时器起时之后，会调用Campaign()方法会将当前节点切换成Candidate状态（或是PerCandidate状态），
	// 底层就是通过发送MsgHup消息实现的
	Campaign(ctx context.Context) error
	// 接收到Client发来的写请求时，Node实例会调用Propose()方法进行处理
	// ，底层就是通过发送MsgProp 消息实现的，
	Propose(ctx context.Context, data []byte) error
	// Client除了会发送读写请求，还会发送修改集群配置的请求（例如，新增集群中的节点），
	// 这种请求Node实例会调用ProposeConfChange（）方法进行处理，底层就是通过发送MsgProp 消息实现的，
	// 只不过其中记录的Entry记录是EntryConfChange类型
	ProposeConfChange(ctx context.Context, cc pb.ConfChangeI) error
	// 当前节点收到其他节点的消息时，会通过Step（）方法将消息交给底层封装的raft实例进行处理
	Step(ctx context.Context, msg pb.Message) error
	// Ready()方法返回的是一个Channel，通过该Channel返回的Ready实例中封装了底层raft实例的相关状态数据，
	// 例如，需妥发送到其他节点的消息、交给上层模块的Entry记录，等等。这是etcd-raft模块与上层模块交互的主要方式
	Ready() <-chan Ready
	// 当上层模块处理完从上述Channel中返回的Ready实例之后，
	// 需要调用Advance（）通知底层的etcd-raft模块返回新的Ready实例
	Advance()
	// 在收到集群配置请求时，会通过调用ApplyConfCha口ge（）方法进行处理
	ApplyConfChange(cc pb.ConfChangeI) *pb.ConfStat
	// TransferLeadership()方法用于Leader节点的转移
	TransferLeadership(ctx context.Context, lead, transferee uint64)
	// Readindex （）方法用于处理只读请求
	ReadIndex(ctx context.Context, rctx []byte) error
	// Status()返回当前节点的状态运行状态，
	Status() Status
	// 通过ReportUnreachable（）方法通知底层的raft实例，当前节点无法与指定的节点进行通信
	ReportUnreachable(id uint64)
	// ReportSnapshot（）方法用于通知底层的raft实例上次发送快照的结采
	ReportSnapshot(id uint64, status SnapshotStatus)
	// Stop关闭当前节点
	Stop()
}

type Peer struct {
	ID      uint64
	Context []byte
}

// 当集群中的节点初次启动时会通过StartNode（）函数启动创建对应的node实例和底层的raft实例。
// 在StartNode（）方法中，主要是根据传入的Config配置创建raft实例并初始化raft使用的相关组件
// Peer封装了节点的工D,peers记录当前集群中全部节点的ID
func StartNode(c *Config, peers []Peer) Node {
	if len(peers) == 0 {
		panic("no peers given; use RestartNode instead")
	}
	rn, err := NewRawNode(c)	// 根据Config配置信息创建raft实例
	if err != nil {
		panic(err)
	}
	err = rn.Bootstrap(peers)
	if err != nil {
		c.Logger.Warningf("error occurred during starting a new node: %v", err)
	}

	n := newNode(rn)	// 初始化node实例
	// 启动一个goroutine，其中会根据底层raft的状态及上层模块传递的数据，
	// 协调处理node中各种远远的数据
	go n.run()
	return &n
}

//当集群中的节点重新启动时不会根据Config中的配置信息调用raftaddNode（）方法添加节点，因为这些信息会从Storage中恢复
func RestartNode(c *Config) Node {
	rn, err := NewRawNode(c)	// 创建底层封装的raft实例，该过程与StartNode（）函数相同
	if err != nil {
		panic(err)
	}
	n := newNode(rn)	// 创建node实例
	// ／启动一个gorouti口e，其中会根据底层raft的状态及上层模块传递的数据，
	// 协调处理node中各种通道的数据
	go n.run()
	return &n
}

type msgWithResult struct {
	m      pb.Message
	result chan error
}

// 节点是节点接口
type node struct {
	propc      chan msgWithResult	// 用于接收MsgProp类型的消息。
	recvc      chan pb.Message	// 除MsgProp外的其他类型的消息都是由该通道接收的
	// 当节点收到EntryConfChange类型的Entry记录时，会转换成ConfChange，
	// 并写入该通道中等待处理。在ConfChange中封装了其唯一ID、待处理的节点ID( NodeID字段）及处理类型(Type字段，
	// 例如，ConfChangeAddNode类型表示添加节点）等信息。
	confc      chan pb.ConfChangeV2
	confstatec chan pb.ConfState	// 在ConfState中封装了当前集群中所有节点的ID，该通道用于向上层模块返回ConfState实例。
	readyc     chan Ready	// 该通道用于向上层模块返回Ready实例，即node.Ready（）方法的返回值
	// 当上层模块处理完通过上述readyc通道获取到的Ready实例之后，
	// 会通过node.Advance（）方法向该通道写入信号，从而通知底层raft实例。
	advancec   chan struct{}
	// 用来接收逻辑时钟发出的信号，之后会根据当前节点的角色推进选举计时器和心跳计时器。
	tickc      chan struct{}
	// 当检测到done通道关闭后，在其上阻塞的goroutine会继续执行，井进行相应的关闭操作。
	done       chan struct{}
	// 当node.Stop（）方法被调用时，会向该通道发送信号，
	// 有另一个goroutine会尝试读取该通道中的内容，当读取到信息之后，会关闭done通道。
	stop       chan struct{}
	status     chan chan Status	// 传递的元素是Channel类型，即node.Status（）方法的返回值。
	rn *RawNode
}

// 完成node实例的初始化
func newNode(rn *RawNode) node {
	return node{
		propc:      make(chan msgWithResult),
		recvc:      make(chan pb.Message),
		confc:      make(chan pb.ConfChangeV2),
		confstatec: make(chan pb.ConfState),
		readyc:     make(chan Ready),
		advancec:   make(chan struct{}),
		// 的规范实现，使tickc成为缓冲chan，因此当节点
		// 忙于处理raft消息时，raft节点可以缓冲一些tick。Raft节点空闲时将恢复进程缓冲
		// 滴答声。
		tickc:  make(chan struct{}, 128),	//	带缓冲的chanel
		done:   make(chan struct{}),
		stop:   make(chan struct{}),
		status: make(chan chan Status),
		rn:     rn,
	}
}

func (n *node) Stop() {
	select {
	case n.stop <- struct{}{}:
		// 尚未停止，因此触发它
	case <-n.done:
		// 节点已停止-无需执行任何操作
		return
	}
	// 阻塞，直到run（）确认停止
	<-n.done
}

// node.run（）方法会处理node结构体中封装的全部通道，它也是node的核心
func (n *node) run() {
	var propc chan msgWithResult	// 指向node.propc通道
	var readyc chan Ready	// 指向node.readyc通道
	var advancec chan struct{}	// 指向node.advancec通道
	var rd Ready

	r := n.rn.raft

	lead := None	// 用于记录当前Leader节点

	for {	// 在该循环中完成了对所有相关Channel的处理
		if advancec != nil {
			// 上层模块还在处理上次从readyc通道返回的Ready实例，所以不能继续向readye中写入数据
			readyc = nil
		} else if n.rn.HasReady() {
			rd = n.rn.readyWithoutAccept()	// 创建Ready实例
			readyc = n.readyc
		}

		if lead != r.lead {	// 检测当前的Leader节点是否发生变化
			// 如果当前节点无法确定集群中的Leader节点，则清空propc，此次循环不再处理MsgPropc消息
			if r.hasLeader() {
				if lead == None {
					r.logger.Infof("raft.node: %x elected leader %x at term %d", r.id, r.lead, r.Term)
				} else {
					r.logger.Infof("raft.node: %x changed leader from %x to %x at term %d", r.id, lead, r.lead, r.Term)
				}
				propc = n.propc
			} else {
				r.logger.Infof("raft.node: %x lost leader %x at term %d", r.id, lead, r.Term)
				propc = nil
			}
			lead = r.lead	// 更新leader
		}

		select {
		case pm := <-propc:	// 读取propc通道，获取MsgPropc消息，并交给raft.Step（）方法处理
			m := pm.m
			m.From = r.id
			err := r.Step(m)
			if pm.result != nil {
				pm.result <- err
				close(pm.result)
			}
		case m := <-n.recvc:	// ／读取node.reeve通道，获取消息（非MsgPrope类型），并交给raft.Step （）方法进行处理
			// 如果来自未知节点的响应消息（例如，MsgHeartbeatResp类型消息）则会被过淀，
			if pr := r.prs.Progress[m.From]; pr != nil || !IsResponseMsg(m.Type) {
				r.Step(m)
			}
		case cc := <-n.confc:	// 读取node.confc通道，获取ConfChange实例，下面根据ConfChange中的类型进行分类处理
			_, okBefore := r.prs.Progress[r.id]
			cs := r.applyConfChange(cc)
			if _, okAfter := r.prs.Progress[r.id]; okBefore && !okAfter {
				var found bool
			outer:
				for _, sl := range [][]uint64{cs.Voters, cs.VotersOutgoing} {
					for _, id := range sl {
						if id == r.id {
							found = true
							break outer
						}
					}
				}
				if !found {
					propc = nil
				}
			}
			select {
			// 将当前集群中的节点信息封装成ConfState实例并写入confstatec通道中
			case n.confstatec <- cs:
			case <-n.done:
			}
		case <-n.tickc:	// 逻辑时钟每推进一次，就会向tickc远远写入一个信号
			// Leader节点推进选举计时器，Follower节点推进心跳计时器
			n.rn.Tick()
		case readyc <- rd:	// 将前面创建的Ready实例写入node.readyc通过中，等待上层楼块读取
			n.rn.acceptReady(rd)
			// 将advancec指向node.advancec远远，这样在下次for循环时，就无法继续向上层模块返回
			// Ready实例了（因为readye会被设立为nil，无法向readye通道中写入Ready实例）
			advancec = n.advancec
		case <-advancec:	// 上层模块处理完Ready实例之后，会向advance通道写入信号
			n.rn.Advance(rd)
			rd = Ready{}
			advancec = nil	// 清空advancec，下次for循环处理的过程中，就能向readye通道写入Ready了
		case c := <-n.status:
			c <- getStatus(r)	// 创建Status实例
		case <-n.stop:
			// 当从stop遥远中读取到信号时，会将done通道关闭，阻塞在其上的goroutine可以继续执行
			close(n.done)
			return
		}
	}
}

// 勾号增加此节点的内部逻辑时钟。选举超时
// 和心跳超时以滴答为单位。
func (n *node) Tick() {
	select {
	case n.tickc <- struct{}{}:
	case <-n.done:
	default:
		n.rn.raft.logger.Warningf("%x A tick missed to fire. Node blocks too long!", n.rn.raft.id)
	}
}

func (n *node) Campaign(ctx context.Context) error { return n.step(ctx, pb.Message{Type: pb.MsgHup}) }

func (n *node) Propose(ctx context.Context, data []byte) error {
	return n.stepWait(ctx, pb.Message{Type: pb.MsgProp, Entries: []pb.Entry{{Data: data}}})
}

func (n *node) Step(ctx context.Context, m pb.Message) error {
	// 忽略通过网络接收的意外本地消息
	if IsLocalMsg(m.Type) {
		// TODO:返回错误？
		return nil
	}
	return n.step(ctx, m)
}

func confChangeToMsg(c pb.ConfChangeI) (pb.Message, error) {
	typ, data, err := pb.MarshalConfChange(c)
	if err != nil {
		return pb.Message{}, err
	}
	return pb.Message{Type: pb.MsgProp, Entries: []pb.Entry{{Type: typ, Data: data}}}, nil
}

func (n *node) ProposeConfChange(ctx context.Context, cc pb.ConfChangeI) error {
	msg, err := confChangeToMsg(cc)
	if err != nil {
		return err
	}
	return n.Step(ctx, msg)
}

func (n *node) step(ctx context.Context, m pb.Message) error {
	return n.stepWithWaitOption(ctx, m, false)
}

func (n *node) stepWait(ctx context.Context, m pb.Message) error {
	return n.stepWithWaitOption(ctx, m, true)
}

// 步骤使用msgs推进状态机。ctx。将返回Err（），
// 如果有。
func (n *node) stepWithWaitOption(ctx context.Context, m pb.Message, wait bool) error {
	if m.Type != pb.MsgProp {
		select {
		case n.recvc <- m:	// 将消息写入reeve或prope通道中
			return nil
		case <-ctx.Done():
			return ctx.Err()
		case <-n.done:
			return ErrStopped
		}
	}
	ch := n.propc
	pm := msgWithResult{m: m}
	if wait {
		pm.result = make(chan error, 1)
	}
	select {
	case ch <- pm:
		if !wait {
			return nil
		}
	case <-ctx.Done():
		return ctx.Err()
	case <-n.done:
		return ErrStopped
	}
	select {
	case err := <-pm.result:
		if err != nil {
			return err
		}
	case <-ctx.Done():
		return ctx.Err()
	case <-n.done:
		return ErrStopped
	}
	return nil
}

func (n *node) Ready() <-chan Ready { return n.readyc }

// 在向readyc通道写入Ready实例之后，会使用pre＊变量记录Ready实例中的相关状态信息，
// 在下次创建Ready实例时，会根据这些pre*变量判断当前底层的raft实例与上次相比是否已经发生了变化，
// 并由此决定是否设置Ready实例的相应字段。
// 当上层模块通过readyec通道读取Ready实例之后，会将其中封装数据进行一系列处理，
// 例如，上层模块会应用Ready.CommittedEntries中的Entry记录、持久化Ready.Entries中的记录和快照数据等。
// 当上层模块处理完此次返回的Ready实例之后，会通过node.Advance（）方法向node.advancec通道写入信号，
// 通知etcd-raft模块其此次返回的Ready实例己经被处理完成，
func (n *node) Advance() {
	select {
	case n.advancec <- struct{}{}: 	// 向advancec通道写入信号
	case <-n.done:	// 当前node己关闭
	}
}

func (n *node) ApplyConfChange(cc pb.ConfChangeI) *pb.ConfState {
	var cs pb.ConfState
	select {
	case n.confc <- cc.AsV2():
	case <-n.done:
	}
	select {
	case cs = <-n.confstatec:
	case <-n.done:
	}
	return &cs
}

// node对外提供了Sta阳的方法，该方法用于读取当前节点的状态
func (n *node) Status() Status {
	c := make(chan Status)	// 创建一个用于传递Status实例的状态
	select {
	case n.status <- c:	// 将该遥远写入到node.status通道中
		return <-c	// 阻塞等待run（）方法写入Status实例
	case <-n.done:
		return Status{}
	}
}

// 当Leader节点发送MsgSnap消息时，如果出现异常，
//则会通过Ra丘接口的ReportUmeaehable（）方法和ReportSnapshot（）方法发送MsgUnreaehable消息和MsgSnapStatus消息
// 在node.ReportUnreachable（）和ReportSnapshot（）方法中，分别会创建MsgUnreachable和MsgSnapStatus消息，
// 然后直接写入node.reeve通道中等待处理，
func (n *node) ReportUnreachable(id uint64) {
	select {
	case n.recvc <- pb.Message{Type: pb.MsgUnreachable, From: id}:
	case <-n.done:
	}
}

// 当Leader节点发送MsgSnap消息时，如果出现异常，
// 则会通过Ra丘接口的ReportUmeaehable（）方法和ReportSnapshot（）方法发送MsgUnreaehable消息和MsgSnapStatus消息
// 在node.ReportUnreachable（）和ReportSnapshot（）方法中，分别会创建MsgUnreachable和MsgSnapStatus消息，
// 然后直接写入node.reeve通道中等待处理，
func (n *node) ReportSnapshot(id uint64, status SnapshotStatus) {
	rej := status == SnapshotFailure	// 用于t主直MsgS口apStatus消息的Reject

	select {
	case n.recvc <- pb.Message{Type: pb.MsgSnapStatus, From: id, Reject: rej}:
	case <-n.done:
	}
}

func (n *node) TransferLeadership(ctx context.Context, lead, transferee uint64) {
	select {
	// 手动设置'from'和'to'，以便leader可以自动转移其leadership 
	case n.recvc <- pb.Message{Type: pb.MsgTransferLeader, From: transferee, To: lead}:
	case <-n.done:
	case <-ctx.Done():
	}
}

func (n *node) ReadIndex(ctx context.Context, rctx []byte) error {
	return n.step(ctx, pb.Message{Type: pb.MsgReadIndex, Entries: []pb.Entry{{Data: rctx}}})
}

// 会读取底层raft实例中的各项数据及相关状态，并最终封装成Ready实例，该Ready实例最终会返回给上层模块，
// 解析一些该函数中几个参数的含义：prevSoftSt、prevHardSt是上次创建Ready实例时记录的raft实例的相关状态
func newReady(r *raft, prevSoftSt *SoftState, prevHardSt pb.HardState) Ready {
	rd := Ready{
		// 获取raftLog中unstable部分存储的Entry记录，这些Entry记录会交给上层楼块进行持久化
		Entries:          r.raftLog.unstableEntries(),
		// 获取己提交但未应用的Entry记录，即raftLog中applied～committed之间的所有记录
		CommittedEntries: r.raftLog.nextEnts(),
		// 获取待发送的消息,最终所有待发送的消息都会记录到raft实例的msgs字段中
		Messages:         r.msgs,
	}
	// 检测两次创建Ready实例之间，raft实例状态是否发生变化，如果无变化，
	// 则将Ready实例相关字段设置为nil，表示无须上层模块处理，
	if softSt := r.softState(); !softSt.equal(prevSoftSt) {
		rd.SoftState = softSt
	}
	if hardSt := r.hardState(); !isHardStateEqual(hardSt, prevHardSt) {
		rd.HardState = hardSt
	}
	// 检测unstable中是否记录了新的快！照数据，如采有，如l将其封装到Ready实例中，交给上层模块进行处理
	if r.raftLog.unstable.snapshot != nil {
		rd.Snapshot = *r.raftLog.unstable.snapshot
	}
	// raft最终会将能响应的请求信息封装成ReadState并记录圭1]readStates中，
	// 这里会检测raft.readStates字段，并将其封装到Ready实例返回给上层模块
	if len(r.readStates) != 0 {
		rd.ReadStates = r.readStates
	}
	rd.MustSync = MustSync(r.hardState(), prevHardSt, len(rd.Entries))
	return rd
}

// 如果Raft条目的硬状态和计数指示
// 需要同步写入持久存储，则MustSync返回true。
func MustSync(st, prevst pb.HardState, entsnum int) bool {
	// 所有服务器上的持久状态：
	// （在响应RPC之前在稳定存储上更新）
	// 当前术语
	// 为
	// 日志条目[]
	return entsnum != 0 || st.Vote != prevst.Vote || st.Term != prevst.Term
}
